package com.liuyu.common.framework.rocketmq.listener;

import com.liuyu.common.framework.rocketmq.kits.RocketConstants;
import com.liuyu.common.framework.rocketmq.strategy.context.RocketTransactionContext;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;

/**
 * rocketmq 事务消息
 * @author: liuyu
 * @date: 2021/11/10 15:30
 * @version: 1
 */
@Slf4j
@RocketMQTransactionListener
public class RocketMqTransactionListenerImpl implements RocketMQLocalTransactionListener {

    private static final String TRANSACTION_CHECK_TIMES = "TRANSACTION_CHECK_TIMES";
    @Autowired
    private RocketTransactionContext rocketTransactionContext;

    /**
     * 执行业务逻辑额功能
     * @author liuyu
     * @date 2021/12/23 10:17
     * @param message
     * @param o
     * @return org.apache.rocketmq.spring.core.RocketMQLocalTransactionState
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        boolean result = rocketTransactionContext.executeTransaction(message, o);
        //如果这里执行成功，则直接进行提交
        if (result) {
            if (log.isDebugEnabled()) {
                String transId = (String) message.getHeaders().get(RocketMQHeaders.PREFIX + RocketMQHeaders.TRANSACTION_ID);
                log.debug("事务消息：{}提交", transId);
            }
            return RocketMQLocalTransactionState.COMMIT;
        }
        return RocketMQLocalTransactionState.UNKNOWN;
    }


    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {

        String transId = (String) message.getHeaders().get(RocketMQHeaders.PREFIX + RocketMQHeaders.TRANSACTION_ID);

        Integer result = rocketTransactionContext.checkTransaction(message);
        //成功，则提交
        if (result == RocketConstants.RocketTransactionState.SUCCESS) {
            if (log.isDebugEnabled()) {
                log.debug("事务消息：{} 本地校验成功", transId);
            }
            return RocketMQLocalTransactionState.COMMIT;
        } else if (result == RocketConstants.RocketTransactionState.ROLLBACK) {
            if (log.isDebugEnabled()) {
                log.debug("事务消息：{} 手动回滚", transId);
            }
            return RocketMQLocalTransactionState.ROLLBACK;
        }
        //未成功，且未达到最大重试次数，继续
        Integer tryTimes = Integer.valueOf((String) message.getHeaders().get(TRANSACTION_CHECK_TIMES));
        if (tryTimes < rocketTransactionContext.getMaxTryTimes(message)) {
            if (log.isDebugEnabled()) {
                log.debug("事务消息：{} 本地校验失败,重试次数：{}", transId, tryTimes);
            }
            return RocketMQLocalTransactionState.UNKNOWN;
        } else {
            //未成功，且已经达到了最大重试次数，回滚
            if (log.isDebugEnabled()) {
                log.debug("事务消息：{} 本地校验失败回滚,重试次数：{}", transId, tryTimes);
            }
            rocketTransactionContext.checkTransactionFail(message);
            return RocketMQLocalTransactionState.ROLLBACK;
        }

    }
}
